package xubo.wangcaifeng.love.CommonFriends
import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object CommonFriends {
  Logger.getLogger("org").setLevel(Level.WARN)
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf()
      .setAppName("共同好友")
      .setMaster("local[*]")
    val sc = new SparkContext(sparkConf)

    /**
      * 想要构建一个图，就得有点，有边，这两个都是RDD
      */
    //rdd[点的数据]   seq这是一个有序的集合
    val userVertex: RDD[(VertexId, (String, Int, String))] = sc.parallelize(Seq(
      (100, ("徐波", 21, "未知")),
      (2, ("彩峰", 25, "未知")),
      (6, ("晓峰", 19, "未知")),
      (9, ("振先", 25, "未知")),
      (133, ("振花", 18, "未知")),

      (16, ("晶巾", 30, "未知")),
      (44, ("振奋", 31, "未知")),
      (21, ("振国", 27, "未知")),
      (138, ("杨治", 28, "未知")),

      (158, ("杨猛", 38, "未知")),
      (5, ("杨勇", 24, "未知")),
      (7, ("宝山", 18, "未知"))
    ))
    //RDD边的数据  Edge这是一个类
    val userEdges: RDD[Edge[Long]] = sc.parallelize(Seq(
      Edge(100, 133, 0),
      Edge(2, 133, 0),
      Edge(9, 133, 0),
      Edge(6, 133, 0),

      Edge(6, 138, 0),
      Edge(16, 138, 0),
      Edge(44, 138, 0),
      Edge(21, 138, 0),

      Edge(5, 158, 0),
      Edge(7, 158, 0)
    ))

    //边和点结合起来    即图对象
    val graph = Graph(userVertex,userEdges)
    //构建好连通图，调用里面的API 获取图中可以连通的分支（connectedComponents()）  vertices是将同一联通图分支中所有的点，向最小的点聚合（vertexId,commonMinId）
    //(1,2)(3,2).....
    val cc: RDD[(VertexId,VertexId)] = graph.connectedComponents().vertices
    //cc.foreach(println)
    //得到这些点是人，所有要的string
    //cc.map(t=>(t._2,t._1.toString)).reduceByKey((a,b)=>a+","+b).foreach(println)

    //现在的需求是将所有人的信息显示出来   想要显示出来就得把用户的信息和连通图jion
    userVertex.join(cc).map{
      case (userId,((name,age,sex),commonMinId)) =>(commonMinId,List((name,age,sex)))
    }.reduceByKey(_++_).foreach(println)


  sc.stop()

  }

}
